package edu.upenn.seas.peerreview.smui
import edu.upenn.seas.peerreview.{Id, Source, Sink}
import scala.actors.Actor._
import scala.actors.Actor

object RPC {
  type dataT = Any
  type dataTList = List[dataT]
  private var seqnum = 0
  val nullSym = Symbol("")
  
  // Save the name of the person you will audit.
  var goingToAudit: Id = null

  /* =object RPC======= Actor=>Agent ===================== */

  //a map from actors to their creating agents
  var actorAgentMap: Map[Actor, RPCAgent[_]] = Map()
  //todo better access control than public (rpc package?)

  def currentAgent = actorAgentMap(self)

  /* =object RPC======= Helper methods ===================== */

  /** given a list and a predicate: if the list contains an element that
   * satisfies the predicate, return the first satisfying element and the
   * list with that element removed, otherwise return
   * None and the original list */
  def removeFirst[A](l: List[A], doRemove: A => Boolean): (Option[A], List[A]) = {
    l match {
      case h :: t if doRemove(h) => { (Some(h), t) }
      case h :: t => removeFirst(t, doRemove) match {
        case (answer, rest) => (answer, h :: rest)
      }
      case Nil => (None, Nil)
    }
  }

  /** Perform an action in a dedicated actor,
   * and wait for the action to complete.
   *
   * I'm not entirely sure this is necessary?*/
  def doInActor(action: => Unit) {
    actor {
      receive {
        case () => action; reply
      }
    } !? ()
  }

  /* =object RPC======= Case classes (sent over transport) ========= */

  /** A payload (either a request or response) carried in an RPCMessage */
  abstract class RPCPayload

  /** A remote function call request or response
   * @param functionname name of function called
   * @param sequence a unique (w.r.t. caller) integer for any call/response pair
   * @param data a RPCRequest or RPCResponse
   */
  case class RPCMessage(functionname: Symbol, sequence: Int,
      data: RPCPayload) extends Message {
    
    /** create a response message with the given return value
     * for this request message; fails if this message is not a request */
    def responseMsg(ret: dataT) = data match {
      case t: RPCRequest => RPCMessage(functionname, sequence, RPCResponse(ret))
      case _ => throw new Exception("can't respond to a response")
    }
    
    /** checks whether the given response message is a response to
     * this request (does not check ids); returns false if this message
     * is not a request
     *
     * msg.checkResponse(msg.responseMsg(anything)) should equal true
     * if msg is a request */
    def checkResponse(response: RPCMessage) = {
      data.isInstanceOf[RPCRequest] &&
      response.data.isInstanceOf[RPCResponse] &&
      (functionname == response.functionname) &&
      (sequence == response.sequence)
    }
  }
  
  //RPCPayloads
  // call a function
  case class RPCRequest(args: dataTList) extends RPCPayload
  // return a value
  case class RPCResponse(ret: dataT) extends RPCPayload

  /** returns function to create a new request message for the given function and arguments */
  def requestMsgFactory = new ((Symbol, dataTList) => RPCMessage) {
    var seqnum = 0
    /** creates a new request message for the given function and arguments */
    def apply(functionname: Symbol, args: dataTList) = {
      seqnum += 1
      RPCMessage(functionname, seqnum, RPCRequest(args))
    }
  }

  /* =object RPC======= Internal case classes ===================== */

  /** An (Id, RPCMessage) pair representing a message from a particular id.
   * Never actually sent over transport, wrapped client-side */
  case class RPCMessageFrom(from: Id, msg: RPCMessage) {
    /** Returns whether this pair is a response to the given
     * request sent to the given id */
    def isResponseTo(to: Id, request: RPCMessage) = 
      (to == from) && request.checkResponse(msg)
  }

  /** A blocked actor, and a function RPCMessageFrom => Boolean indicating
   * what messages can wake this actor up */
  case class BlockedActor(canAccept: RPCMessageFrom => Boolean, actor: Actor)

  /** sent by actor to unblock the handle() method */
  case class ThreadYield(wakeMeUpOn: RPCMessageFrom => Boolean)

  /** sent by actor to unblock the handle() method and
   * remove this actor from the list of running actors */
  case object ThreadExit

  /** sent by handler to kill this thread */
  case object Kill

}
import RPC._


/* =========  RPC object interfaces ===================== */

// any RPC-callable interface extends this 
trait RPCInterface {
  def default[T](args: dataT*): T 
}

// server-side handler. by default, will handle no methods unless overriden.
class RPCHandler {
  def default[T](args: dataT*): T =
    throw new UnsupportedOperationException()
}
// a proxy for a remote ui/sm
class RPCProxy(val id: Id) {
  def default[T](args: dataT*): T = {
    val caller = new Throwable().getStackTrace()(1)
    val classname = caller.getClassName()
    val methname = caller.getMethodName()
    //println("Proxy: Calling %s.%s(%s)".format(classname,methname,args.mkString(",")))
    currentAgent.makeCallWArgs(id, Symbol(methname), args.toList).asInstanceOf[T]
  }
}

trait RPCThread {
  def join()
  def kill()
}

case class RPCId[IFace <: RPCInterface](id: Id)

/* ========= RPC agent class (Source.T) ===================== */

abstract class RPCAgent[IFace <: RPCInterface](myId:Id, transport: Sink.T) extends Source.T {
  
  protected val _debug=edu.upenn.seas.peerreview.Debug.debug("actors")(_)

  var debugPrintLevel = 0
  def debug(s: String) = {_debug(" "*debugPrintLevel*5 + myId + " " + s)}
  def debugEnter(s: String) = {debug(s); debugPrintLevel += 1; }
  def debugLeave(s: String) = {debugPrintLevel -= 1; debug(s); }
        

  /* =========== Internal methods */

  val requestMsg = requestMsgFactory

  /** The parent actor; blocked while a thread is executing
   * (send ThreadYield or ThreadExit to unblock).
   * Typically the dispatcher, but may be the parent thread during a fork. */
  private var parentActor: scala.actors.OutputChannel[Any] = null

  private var blockedThreads: List[BlockedActor] = Nil
  private var msgQ: List[RPCMessageFrom] = Nil

  /** Returns whether any waiting threads can accept the given message */
  def anyThreadAccept(msg: RPCMessageFrom) = 
    blockedThreads.find(_.canAccept(msg)) != None

  // To fulfill the Source.T: Incoming message.
  def apply(from : Id, message : Message) {
    message match {

      case m: RPCMessage => 
        handleIncomingRPCMessage(RPCMessageFrom(from, m))

      case AuditResult(r) => 
        println("[%s] RPC audit: %s was %s".format(myId, goingToAudit, if (r) "not cheating." else "cheating?"))

      case x => {
        new Exception("Didn't get an RPCenvelope. Cry. " + x).printStackTrace()
        None // Don't throw for now.
        }
    }
  }

  /** add a thread to the blocked thread list, to be woken up when a message
   * that satisfys the given function arrives */
  private def addBlockedThread(wakeup: RPCMessageFrom=> Boolean, thr: Actor) {
    blockedThreads = BlockedActor(wakeup, thr) :: blockedThreads
  }

  /** send the given message to the given actor, and block until the actor
   * yields or exits. If it yielded (with a wakeup function), add it to the
   * blocked threads list. */
  private def blockingCallToBlockedThread(thr: Actor, msg: Any) {
    debugEnter("Dispatcher block, %x ! %s".format(thr.hashCode, msg))
    doInActor {
      thr ! msg
      receive {
        case ThreadYield(wakeup) => addBlockedThread(wakeup, thr)
        case ThreadExit =>
      }
    }
    debugLeave("Dispatcher unblock")
  }

  /** Dispatch a message to the appropriate blocked thread, updating the
   * blocked thread list as necessary. Assumes that some thread can accept the
   * message. */
  private def handleRPCMessage(msg: RPCMessageFrom) {
    debug("%d java thr count; %d Blocked threads; %d waiting messages: %s".format(Thread.activeCount, blockedThreads.size, msgQ.size, msgQ))
    def iCanAcceptMsg(a: BlockedActor) = a.canAccept(msg)
    removeFirst(blockedThreads, iCanAcceptMsg _) match {
      case (Some(BlockedActor(canAccept, thr)), blockedThreadsRest) => {
        blockedThreads = blockedThreadsRest
        blockingCallToBlockedThread(thr, msg)
      }
      case (None, _) => throw new Exception("message "+msg+" can't be handled")
    }
  }

  /** Handles a message sent to the Source.T handle method.
   *
   * Effectively: Adds the message to end of the queue, and repeatedly
   * dispatches the first message on the queue that can be handled until no messages
   * on the queue can be handled. */
  private def handleIncomingRPCMessage(msg: RPCMessageFrom) {
    if (!anyThreadAccept(msg)) {
      //no one accepted it, so enqueue it and return
      msgQ = msgQ ::: List(msg)
      return
    }

    handleRPCMessage(msg)
    while (true) {
      removeFirst(msgQ, anyThreadAccept _) match {
        case (Some(msg), msgQRest) => {
          assert((msgQRest.size + 1) == msgQ.size)
          msgQ = msgQRest
          handleRPCMessage(msg)
        }
        case (None, _) => return
      }
    }
  }

  /** Send a message to the parent actor. */
  private def alertParent(m: Any) { parentActor ! m }

  /** Block the current thread until a message satisfying the given predicate
   * arrives, set the parent/dispatcher to the sender of the message, and return
   *  that message. Does not unblock the previous parent/dispatcher. */
  private def blockUntilReceive(wakeup: RPCMessageFrom => Boolean): RPCMessageFrom = {
    debug("Thread block %x".format(self.hashCode))
    val res = receive {
      case Kill => exit //send threadexit?
      case m: RPCMessageFrom => parentActor = sender; m
    }
    debug("Thread unblock %x".format(self.hashCode))
    res
  }

  /** Same as blockUntilReceive, but first unblocks the parent/dispatcher. */
  private def yieldAndBlockUntilReceive(wakeup: RPCMessageFrom => Boolean): RPCMessageFrom = {
    alertParent(ThreadYield(wakeup))
    blockUntilReceive(wakeup)
  }
  
  /* ========== Main method =======*/

  private val mainActor = fork { main() }

  // main is started thru msg passing, to avoid 120AdventureExtensionProblem
  send(myId, msgQ.head.msg)
  msgQ = msgQ.tail

  /* ========== Interface for overriding classes =======*/

  //should be protected, + visible to rpc package
  /** Perform a blocking RPC call: to.functionname(args), and return the result. */
  def makeCallWArgs(to: Id, functionname: Symbol, args: dataTList): dataT = {
    val request = requestMsg(functionname, args)
    send(to, request)

    yieldAndBlockUntilReceive(_.isResponseTo(to, request)) match {
      case RPCMessageFrom(from, RPCMessage(_,_,RPCResponse(ret))) => ret
      case m => throw new Exception("expected RPCResponse, got "+m)
    }
  }

  /** Given a partial function (fromId, functionname, args)=>ret, block the
   * current thread (once) until an RPC call arives that the function can
   * handle, and then call the function and send back the return value. */
  protected def respondToCall(callHandler: PartialFunction[(Id, Symbol, dataTList), dataT]) {
    def accept(r: RPCMessageFrom) = r match {
      case RPCMessageFrom(from, RPCMessage(functionname, _, RPCRequest(args))) =>
        callHandler.isDefinedAt(from, functionname, args)
      case _ => false
    }

    val req = yieldAndBlockUntilReceive(accept _)
    req match {
      case RPCMessageFrom(from, RPCMessage(functionname, _, RPCRequest(args))) => 
        send(from, req.msg.responseMsg(callHandler(from, functionname, args)))
    } 
  }

  protected def fork(childfunc: =>Unit) = {

    // has unique sequence number
    val wakeupMsg = RPCMessageFrom(myId, requestMsg(Symbol(""), Nil))

    val newThread = actor {
      debug("fork! %x".format(self.hashCode))
      blockUntilReceive(_ == wakeupMsg)
      childfunc
      alertParent(ThreadExit)
    }

    actorAgentMap += newThread -> this

    addBlockedThread(_ == wakeupMsg, newThread)

    debug("generating wakeup for %x".format(newThread.hashCode))
    msgQ = msgQ ::: List(wakeupMsg)

    new RPCThread() {
      private val myActor = newThread
      private var finished = false
      def kill() {
        myActor ! Kill
        actorAgentMap -= myActor
        blockedThreads = blockedThreads.filter(_.actor != myActor)
      }
      def join() { throw new Exception("join not implemented") }
    }
  }

  /** The main thread of control for an agent */
  def main()

  /** Send a message to the transport layer. */
  final def send(to: Id, m: Message) { 
    // Intercept to catch the auditee.
    m match {
      case PleaseAudit(opp: Id) => goingToAudit = opp
      case _ => 
    }
    transport(to, m) 
  }

  /* ============= helper functions for overriding classes ======= */

  private var _RPCSender: Id = null
  protected def RPCSender = _RPCSender

  protected def forkhandler(functionname: Symbol)(f: Seq[Any]=>Any) = fork {
    while (true) {
      respondToCall {
        case (from, sym, data) if sym == functionname =>
          _RPCSender = from
          f(data.asInstanceOf[Seq[Any]])
      }
    }
  }
  
  protected def makeCall(to: Id, functionname: Symbol, args: dataT*): dataT = {
    makeCallWArgs(to, functionname, args.toList)
  }
}
